use crate::{
    cmd, config,
    data::{self, VCData},
    file,
    leader::{FastElectingRule, FastVotingRule},
    ws::{WsEvent, WsResponse},
};
use actix::prelude::*;
use rand::{rngs::ThreadRng, Rng};
use std::{
    collections::HashMap,
    fmt::Debug,
    time::{Duration, Instant},
};
// https://github.com/actix/examples/blob/master/websockets/chat/src/server.rs
// https://cloud.tencent.com/developer/article/1756850

/// Chat server sends this messages to session
#[derive(Message)]
#[rtype(result = "()")]
pub struct Message(pub String);

// #[derive(Message)]
// #[rtype(result = "()")]
// pub struct Binary(pub Bytes);

/// Message for chat server communications

/// New chat session is created
#[derive(Message, Debug)]
#[rtype(usize)]
pub struct Connect {
    // pub id: usize,
    // pub myid: usize,
    pub client_id: usize,
    pub addr: Recipient<Message>,
}

/// Session is disconnected
#[derive(Message)]
#[rtype(result = "()")]
pub struct Disconnect {
    pub session_id: usize,
    pub myid: usize,
    pub client_id: usize,
    // pub namespace: String,
    /// Is closed connection?
    pub closed: bool,
    // pub ip: IpAddr,
}

/// Send message to specific cluster
// #[derive(Message)]
// #[rtype(result = "()")]
// pub struct ClientMessage {
//     /// Id of the client session
//     pub id: usize,
//     /// Peer message
//     pub msg: String,
//     /// cluster name
//     pub cluster: String,
// }

#[derive(Message)]
#[rtype(result = "()")]
pub struct Broadcast;

// #[derive(Message, Default, Debug)]
// #[rtype(result = "()")]
// pub struct FsBroadcast {
//     // Create File
//     // Create Directory
//     pub c: Option<Vec<PathBuf>>,
//     pub cd: Option<Vec<PathBuf>>,

//     // Remove File
//     // Remove Directory
//     pub r: Option<Vec<PathBuf>>,
//     pub rd: Option<Vec<PathBuf>>,

//     // Rename=> from:to
//     pub x: Option<HashMap<PathBuf, PathBuf>>,

//     // DataChange
//     pub d: Option<Vec<PathBuf>>,

//     // Metadata accessed
//     // Metadata permissions
//     // Metadata modified
//     pub a: Option<Vec<PathBuf>>,
//     pub p: Option<Vec<PathBuf>>,
//     pub m: Option<Vec<PathBuf>>,
// }

// 数据同步
// #[derive(Message)]
// #[rtype(Copy)]
// pub struct Copy {
//     pub session_id: usize,
//     pub vcd: VCData,
// }

// 数据同步
#[derive(Message)]
#[rtype(result = "()")]
pub struct ResetVote;

// 节点内部通讯
#[derive(Message)]
#[rtype(result = "()")]
pub struct Heartbeat;

// 投票
#[derive(Debug, Message)]
#[rtype(Election)]
pub struct Election {
    // pub session_id: usize,
    // 实际数据
    pub vcd: VCData,
    // 锁等待超时
    // pub timeout: bool,
    // 投票结果
    pub result: Option<VotingResult>,
}

impl Election {
    fn pollout(&mut self, id: usize, p: usize) {
        self.vcd.pollout(id, p);
    }

    fn sync(&mut self, vcd: &VCData) {
        self.vcd.sync_with(vcd);
    }

    pub fn fmt(&self) -> String {
        format!("{{ result = {:?}, vcd = {} }}", self.result, self.vcd.fmt())
    }
}

#[derive(Debug)]
pub enum VotingResult {
    // 投票成功
    Ok,
    // 投票失败
    Failed,
    // 刚加入的节点,已经有leader了
    Leader,
}

/// `ChatServer` manages chat rooms and responsible for coordinating chat session.
///
/// Implementation is very naïve.
#[derive(Debug)]
pub struct ChatServer {
    // SESSION_ID, ADDR
    sessions: HashMap<usize, Recipient<Message>>,
    rng: ThreadRng,
    // SESSION_ID, VOTER_ID
    session_keys: HashMap<usize, usize>,
    // 连接失败数统计，超过3次，则降级
    // hb_failed_count: usize,
    // leader 下线的次数统计
    // lost_leader_count: usize,
    // leader 是否已经加入
    has_leader: Option<bool>,
    // 投票规则
    voting_rule: FastVotingRule,
    // 选举规则
    electing_rule: FastElectingRule,
    // 上次广播时间: SRV_NAME: Duration
    last_broadcasts: HashMap<String, Instant>,
}

impl ChatServer {
    pub fn new() -> ChatServer {
        let mut last_broadcasts = HashMap::new();
        for srv_name in config::get_registry_services().keys() {
            last_broadcasts.insert(srv_name.to_owned(), Instant::now());
        }
        ChatServer {
            sessions: HashMap::new(),
            rng: rand::thread_rng(),
            session_keys: HashMap::new(),
            // hb_failed_count: 0,
            // lost_leader_count: 0,
            has_leader: None,
            voting_rule: FastVotingRule::new(),
            electing_rule: FastElectingRule::new(),
            last_broadcasts,
        }
    }
}

impl ChatServer {
    fn broadcast(&mut self, msg: WsResponse) {
        // 广播给其他节点
        let myid = config::get_server_id();
        for (sid, addr) in self.sessions.iter() {
            if let Some(vid) = self.session_keys.get(&sid) {
                if *vid == myid {
                    continue;
                }
                log::info!("Broadcasting to {vid} session {sid}");
                addr.do_send(Message(msg.to_string()));
            }
        }
    }

    fn broadcast_all(&mut self, msg: WsResponse) {
        // 广播给其他节点
        for (sid, addr) in self.sessions.iter() {
            if let Some(vid) = self.session_keys.get(&sid) {
                log::info!("Broadcasting to {vid} session {sid}");
                addr.do_send(Message(msg.to_string()));
            }
        }
    }

    fn broadcast_data(&mut self, data: HashMap<String, HashMap<String, String>>) {
        // 广播给其他节点
        for (sid, addr) in self.sessions.iter() {
            if let Some(vid) = self.session_keys.get(&sid) {
                log::debug!("Broadcasting to {vid} session {sid}");

                let vcd;
                /*Mutex*/
                {
                    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                    let mut loc = lock.lock();
                    // Acquired Lock: ReadOnly
                    loc.vcd.tranx += 1;
                    vcd = loc.vcd.clone();
                    cvar.notify_all();
                    // Released Lock
                }

                let mut msg =
                    WsResponse::ok(Some(WsEvent::BF), "Broadcast file".to_owned(), Some(vcd));
                msg.dat = Some(data.clone());
                addr.do_send(Message(msg.to_string()));
            }
        }
    }
}

/// Make actor from `ChatServer`
impl Actor for ChatServer {
    type Context = Context<Self>;
}

/// Handler for Connect message.
///
/// Register new session and assign unique id to this session
impl Handler<Connect> for ChatServer {
    type Result = usize;

    // 生成session_id
    fn handle(&mut self, msg: Connect, _: &mut Context<Self>) -> Self::Result {
        let session_id: u64 = self.rng.gen::<u64>();
        let session_id = session_id as usize;
        self.sessions.insert(session_id, msg.addr.clone());
        self.session_keys.insert(session_id, msg.client_id);
        log::info!(
            "Session created for {}, session ID is {}",
            msg.client_id,
            session_id
        );

        /*Mutex*/
        {
            let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
            let mut loc = lock.lock();
            // Acquired Lock: ReadWrite
            log::trace!("Voter_id: {}, online", msg.client_id);
            loc.online(msg.client_id);
            cvar.notify_all();
            // Released Lock
        }

        // send id back
        session_id
    }
}

impl Handler<Broadcast> for ChatServer {
    type Result = ();
    fn handle(&mut self, _: Broadcast, _: &mut Context<Self>) {
        /*Mutex*/
        {
            let &(ref lock, ref _cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
            let loc = lock.lock();
            // Acquired Lock: ReadOnly
            if !loc.is_leading() {
                // 不是leading，不广播
                return;
            }
            // Released Lock
        }

        // 全部执行完成后，一次性广播
        // 与上次的时间比较，如果超过执行的时间，则推送数据
        let mut data = HashMap::new();

        for (srv_name, en) in config::get_registry_services() {
            // enable_broadcast => false
            if let Some(enable_broadcast) = en.enable_broadcast {
                if !enable_broadcast {
                    continue;
                }
            } else {
                continue;
            }
            let rbe = match en.broadcast {
                Some(broadcast) => broadcast,
                None => continue,
            };
            // enable_broadcast => false ... end

            let intv = self.last_broadcasts.get(&srv_name).unwrap();
            let timeout = Duration::from_secs(rbe.interval_secs.unwrap_or(3) as u64);
            let now = Instant::now();
            if now.duration_since(*intv) >= timeout {
                self.last_broadcasts.insert(srv_name.clone(), now);

                // 广播事件
                let stdout = match cmd::run_cmd(
                    &srv_name,
                    rbe.cmd,
                    if rbe.root.is_some() {
                        rbe.root
                    } else {
                        en.root.clone()
                    },
                    None,
                    en.env,
                ) {
                    Some((stdout, _stderr)) => stdout,
                    None => continue,
                };

                let mut data2: HashMap<String, String> = HashMap::new();
                for mut line in stdout.lines() {
                    if line.starts_with("export ") {
                        let (_first, last) = line.split_at(7);
                        line = last;
                    }

                    let values = line.split_once("=");
                    if let Some(values) = values {
                        // 保存key=value
                        let prop = values.0.to_owned();
                        if !rbe.props.contains(&prop) {
                            continue;
                        }
                        data2.insert(prop, values.1.to_owned());
                    }
                }
                // 写入广播文件，与其他节点保持一致
                if let Some(output) = rbe.output {
                    file::write_broadcast_output(&srv_name, &output, en.root, &data2);
                }

                data.insert(srv_name, data2);
            }
        }

        self.broadcast_data(data);
    }
}

// 清理投票信息，重新投票
impl Handler<ResetVote> for ChatServer {
    type Result = ();
    fn handle(&mut self, _: ResetVote, _: &mut Context<Self>) {
        let myvcd;
        /*Mutex*/
        {
            let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
            let mut loc = lock.lock();
            // Acquired Lock: ReadWrite
            myvcd = loc.vcd.clone();
            loc.reset();
            log::info!("Agree to SWITCHOVER, Reset Data");
            cvar.notify_all();
            // Released Lock
        }

        // 广播给其他节点
        self.broadcast_all(WsResponse::ok(
            Some(WsEvent::RV),
            "Broadcast to Reset Data".to_owned(),
            Some(myvcd),
        ));
    }
}

/// 内部心跳
/// 通过WsChatSession::hb调用
impl Handler<Heartbeat> for ChatServer {
    type Result = ();
    fn handle(&mut self, _: Heartbeat, _: &mut Context<Self>) {
        let myid = config::get_server_id();
        let attendees;
        let status;
        let mut cloned_vcd;
        /*Mutex*/
        {
            let &(ref lock, ref _cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
            let loc = lock.lock();
            // Acquired Lock: ReadOnly
            status = loc.status.clone();
            attendees = loc.attendees.len();
            cloned_vcd = loc.vcd.clone();

            log::info!(
                "{:?} in {:?}, Members {:?}, Myid {}, LDR {}",
                loc.role,
                status,
                loc.attendees.keys(),
                myid,
                loc.vcd.leader
            );
            // Released Lock
        }

        // 是否需要降级为follower
        match status {
            data::Status::Looking => {
                if self.electing_rule.electing(attendees, &cloned_vcd) {
                    // 投票周期+1
                    {
                        let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                        let mut loc = lock.lock();
                        // Acquired Lock: ReadWrite
                        loc.vcd.term += 1;
                        let term = loc.vcd.term;
                        // 选举成功
                        loc.sync_leader(myid, term, data::Status::Leading);
                        // 同步给客户端
                        cloned_vcd.sync_with(&loc.vcd);

                        cvar.notify_all();
                        // Released Lock
                    }
                    self.has_leader = Some(true);

                    // 同步给客户端
                    log::info!("Election Ended, Leader is {myid}");
                    // 广播给其他节点
                    self.broadcast(WsResponse::ok(
                        Some(WsEvent::FA),
                        "Broadcast to Voter".to_owned(),
                        Some(cloned_vcd),
                    ));

                    // 切换为leader
                    cmd::on_switch_to_leader();
                }

                return;
            }
            data::Status::Pending => {
                // 过度阶段，这个状态的节点只能晋升为Follower
            }
            data::Status::Leading => {
                // leader状态是否一直有效
                // 避免脑裂
                if !self.electing_rule.is_valid(attendees) {
                    log::warn!(
                        "Lack of quorum in Leading, expect {}, now {}, Status Reset to Looking",
                        self.electing_rule.quorum(),
                        attendees,
                    );
                    {
                        let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                        let mut loc = lock.lock();
                        // Acquired Lock: ReadWrite
                        loc.reset();
                        cvar.notify_all();
                        // Released Lock
                    }

                    cmd::on_switch_to_follower();
                } else {
                    // 更新时间

                }
            }

            data::Status::Following => {
                let mut off = false;
                /*Mutex */
                {
                    let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                    let mut loc = lock.lock();
                    // Acquired Lock: ReadWrite
                    // leader offline
                    if loc.is_leader_offline() {
                        off = true;
                        log::warn!("Unable to connect to Leader, Switch to looking");
                        // 状态改为等待投票
                        loc.reset();
                    }
                    cvar.notify_all();
                    // Released Lock
                }

                if off {
                    // 切换为follower
                    cmd::on_switch_to_follower();
                }
            }
        }
    }
}

// leader竞选
// 1.term大的直接胜出
// 2.term相同，事务id大的胜出
// 3.事务id相同，服务器id大的胜出
impl Handler<Election> for ChatServer {
    type Result = MessageResult<Election>;

    fn handle(&mut self, mut voter: Election, _: &mut Context<Self>) -> Self::Result {
        let myid;
        let status;
        let cloned_vcd;
        let max_weight;
        /* Mutex */
        {
            let &(ref lock, ref _cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
            let loc = lock.lock();
            // Acquired Lock: ReadOnly
            myid = loc.vcd.myid;
            cloned_vcd = loc.get_vcd();
            status = loc.status.clone();
            max_weight = loc.get_max_weight();
            // Released Lock
        }

        log::debug!("Election with {}, Raw <<< {}", voter.vcd.myid, voter.fmt());

        // 如果本节点状态为leading或Following，则无需投票
        match status {
            data::Status::Following | data::Status::Leading => {
                // 只需同步数据
                voter.sync(&cloned_vcd);
                voter.result = Some(VotingResult::Leader);
            }
            data::Status::Pending => {
                // 过度阶段
                voter.result = Some(VotingResult::Failed);
            }
            data::Status::Looking => {
                // Start Voting
                let voting_result = self
                    .voting_rule
                    .voting(&cloned_vcd, &voter.vcd, &max_weight);
                if voting_result {
                    /* Mutex */
                    {
                        let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
                        let mut loc = lock.lock();
                        // Acquired Lock: ReadWrite
                        loc.poll(&voter.vcd);
                        cvar.notify_all();
                        // Released Lock
                    }
                    // Voter已投出
                    voter.pollout(myid, voter.vcd.poll);
                    //
                    voter.result = Some(VotingResult::Ok);
                } else {
                    // 投票失败
                    voter.result = Some(VotingResult::Failed);
                }
            }
        }

        log::debug!(
            "Election with {}, Raw >>> {}",
            voter.vcd.myid,
            voter.fmt()
        );
        return MessageResult(voter);
    }
}

/// Handler for Disconnect message.
impl Handler<Disconnect> for ChatServer {
    type Result = ();

    fn handle(&mut self, msg: Disconnect, _: &mut Context<Self>) {
        if msg.closed || msg.client_id == msg.myid {
            return;
        }
        let myid = msg.myid;
        let leader;
        /* Mutex */
        {
            let &(ref lock, ref cvar) = &*data::SHARE_GLOBAL_AREA_MUTEX_PAIR.clone();
            let mut loc = lock.lock();
            // Acquired Lock: ReadWrite
            leader = loc.vcd.leader;
            loc.offline(msg.client_id);
            cvar.notify_all();
            // Released Lock
        }

        // 从会话列表中删除
        let _removed_vid = match self
            .sessions
            .remove(&msg.session_id)
            .and_then(|_| self.session_keys.remove(&msg.session_id))
        {
            Some(v) => {
                log::info!(
                    "The {} Session {} removed",
                    msg.client_id,
                    msg.session_id,
                );
                v
            }
            None => return,
        };

        // leader异常，则无需打印，且无法广播数据到其他节点
        if myid == leader {
            return;
        }

        // 通知其他会话，排除本地心跳会话
        // session_keys: session_id: client_id
        for (sid, addr) in self.sessions.iter() {
            let vid = match self.session_keys.get(&sid) {
                None => continue,
                Some(id) => id,
            };

            // 本地连接
            if *vid == myid {
                continue;
            }

            // 针对本机使用CTRL+C停止，且本机非leader时，会提示“Leader off-lined, broadcast to...”
            // 解决方式：取消signal：CTRL+C
            let mut offline_node_vcd = VCData::new(0);
            offline_node_vcd.myid = msg.client_id;
            offline_node_vcd.leader = leader;

            let message = if leader == msg.client_id {
                log::info!(
                    "Leader off-lined, broadcast to {vid} of session {sid}",
                );
                WsResponse::ok(
                    Some(WsEvent::LA),
                    "Leader Offline".to_owned(),
                    Some(offline_node_vcd),
                )
            } else {
                log::info!(
                    "Follower off-lined, broadcast to {vid} of session {sid}",
                );

                // 应通知对方那个节点离线了
                WsResponse::ok(
                    Some(WsEvent::TV),
                    "Offline".to_owned(),
                    Some(offline_node_vcd),
                )
            };

            addr.do_send(Message(message.to_string()));
        }
    }
}
